package com.catmiao.rdd.operate.part;

import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.Partitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.List;

/**
 * @author ChengMiao
 * @title: Transfer_01_Map
 * @projectName spark_study
 * @description: 分区器
 * @date 2024/11/25 16:27
 */
public class Part_02 {

    public static void main(String[] args) throws InterruptedException {

        final SparkConf conf = new SparkConf();
        conf.setAppName("appName");
        conf.setMaster("local[*]");

        final JavaSparkContext jsc = new JavaSparkContext(conf);


        List<Tuple2<String,Integer>> datas = new ArrayList<>();
        datas.add(new Tuple2<String, Integer>("nba",1));
        datas.add(new Tuple2<String, Integer>("cba",2));
        datas.add(new Tuple2<String, Integer>("a",3));
        datas.add(new Tuple2<String, Integer>("a",4));
        datas.add(new Tuple2<String, Integer>("cba",4));


        JavaRDD<Tuple2<String, Integer>> rdd = jsc.parallelize(datas, 3);

        JavaPairRDD<String, Integer> pairRDD = rdd.mapToPair(
                kv -> {
                    System.out.println("###############");
                    return kv;
                }
        );


        JavaPairRDD<String, Integer> reduceRdd = pairRDD.reduceByKey(new MyPartitioner(3), Integer::sum);

        reduceRdd.saveAsTextFile("output/part/after2");



        System.out.println("2计算完毕");


        // RDD在foreach循环时，逻辑代码和操作全部都是在Executor端执行的，结果并不会拉取回Driver端

        // TODO 如果Executor端使用了Driver端的数据，那么需要从Driver端拉取数据到Executor端，数据拉取的单位是Task


        /**
         * RDD 不能以Executor为单位进行数据传输，而是以Task为单位
         * 
         * Spark需要采用特殊的数据模型来实现数据传输，这个数据模型就是Broadcast
         */
        

        jsc.close();


    }
}


/**
 * 自定义分区器
 */

class MyPartitioner extends Partitioner {

    private int numPartitions = 1;

    public MyPartitioner(int numPartitions) {
        this.numPartitions = numPartitions;
    }


    // TODO 指定分区数量
    @Override
    public int numPartitions() {
        return numPartitions;
    }

    // TODO 根据数据的key来获取数据的分区编号，从0开始
    @Override
    public int getPartition(Object key) {
        if(ObjectUtils.equals(key,"nba")){
            return 0;
        }else if(ObjectUtils.equals(key,"cba")){
            return 1;
        }

        return 2;
    }
}


